package vn.com.epi.log.Import;

import java.io.IOException;
import java.io.PrintWriter;
import java.io.StringWriter;
import java.util.Collections;
import java.util.List;
import java.util.Map;

import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;

import vn.com.epi.data.table.ArticleTable;
import vn.com.epi.data.table.TempUserTable;
import vn.com.epi.data.table.UserTable;
import vn.com.epi.utils.StringUtils;
import vn.com.epi.utils.TableUtils;

import com.google.common.base.Strings;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;

public class LogConcatenate {
	
	private static final int MAX_BLOCK_SIZE = 1000;
	private static final int TWO_DAYS = 172800;
	
	private static Map<Integer, Integer> category;
	private static int[] categoryCode = {45,47,50,52,53,54,55,58,59,71,72,76,79,82,84,87,88,104,107,108,119,121,122,126,127,132,136,137,138,139,141,144,145,146,147,148,150,152};
	
	private static int getcategoryId(int categoryID) {
		int first = 0;
		int last = categoryCode.length - 1;
		while (first <= last) {
			int mid = (first + last)/2;
			if (categoryCode[mid] == categoryID)
				return mid + 1;
			else
				if (categoryCode[mid] < categoryID)
					first = mid + 1;					
				else
					last = mid - 1;					
		}
		int index = -1;
		return index;
	}		

	private static void loadCategory() throws IOException {
        HTable table = new HTable(ArticleTable.NAME);
        category = Maps.newHashMap();
		try {
			Scan scan = new Scan();
			scan.addColumn(ArticleTable.FAM_MAIN, ArticleTable.COL_CATEGORY_ID);
			ResultScanner rs = table.getScanner(scan);
			for (Result r : rs) {
				int articleId = Bytes.toInt(r.getRow());
				int categoryId = Bytes.toInt(r.getValue(ArticleTable.FAM_MAIN, ArticleTable.COL_CATEGORY_ID));
				categoryId = getcategoryId(categoryId);
				if (categoryId < 0)
					continue;
				category.put(articleId, categoryId);
			}
		} catch (Exception e) {
			System.err.println("Error while load category ");
			e.printStackTrace();
		} finally {
			table.close();
		}
	}

	private static void process(HTable table, Map<String, UserMultiRecord> map) {
		List<Get> gets = Lists.newArrayList();		
		UserMultiRecord[] records = new UserMultiRecord[map.keySet().size()];
		
		try {
			int index = 0;
			for (String userId : map.keySet()) {
				byte[] row = Bytes.toBytes(userId);
				Get get = new Get(row);
				gets.add(get);
				records[index] = map.get(userId);
				index++;
			}
			Result[] results = table.get(gets);
			List<Put> puts = Lists.newArrayList();
			
			for (int i = 0; i < results.length; i++) 
				if (!Strings.isNullOrEmpty(records[i].getArticleIds()))	{
					byte[] row = gets.get(i).getRow();
					String articleIds = Bytes.toString(results[i].getValue(UserTable.FAM_MAIN, UserTable.COL_VISITED_ARTICLES));
					String timestamps = Bytes.toString(results[i].getValue(UserTable.FAM_MAIN, UserTable.COL_VISITED_TIMESTAMPS));
					StringBuilder articleSb = new StringBuilder();
					StringBuilder timestampSb = new StringBuilder();
					if (Strings.isNullOrEmpty(articleIds)) {
						articleSb.append(records[i].getArticleIds());
						timestampSb.append(records[i].getTimestamps());
					} else {
						String[] chunks = articleIds.split(",");
						Map<String, Boolean> visited = Maps.newHashMap();
						for (int j = 0; j < chunks.length; j++) {
							visited.put(chunks[j], true);
						}
						articleSb.append(articleIds).append(",");
						timestampSb.append(timestamps).append(",");
						String[] newArticles = records[i].getArticleIds().split(",");
						String[] newTimestamps = records[i].getTimestamps().split(",");
						for (int j = 0; j < newArticles.length; j++) 
							if (!visited.containsKey(newArticles[j])) {
								articleSb.append(newArticles[j]).append(",");
								timestampSb.append(newTimestamps[j]).append(",");
								visited.put(newArticles[j], true);
							}
						StringUtils.deleteLastChar(articleSb);
						StringUtils.deleteLastChar(timestampSb);
					}
					Put put = new Put(row);          
					put.add(UserTable.FAM_MAIN, UserTable.COL_VISITED_ARTICLES, Bytes.toBytes(articleSb.toString()));
					put.add(UserTable.FAM_MAIN, UserTable.COL_VISITED_TIMESTAMPS, Bytes.toBytes(timestampSb.toString()));
					puts.add(put);
				}
			if (puts.size() > 0)
				table.put(puts);
			
		} catch (Exception e) {
			StringWriter sw = new StringWriter();
			PrintWriter pw = new PrintWriter(sw);
			e.printStackTrace(pw);
			System.err.println("Error while process concat: " + sw.toString());
			System.exit(-1);
		}
	}
	
	private static void cutOff(HTable table, Map<String, UserMultiRecord> map) throws IOException {
		Map<Integer, Integer> distribution;
		List<UserRecord> list = Lists.newArrayList();
		List<Put> puts = Lists.newArrayList();
		List<Delete> deletes = Lists.newArrayList();
		long current = System.currentTimeMillis() / 1000L;
		for (String userId : map.keySet()) {
			UserMultiRecord record = map.get(userId);
			if (!Strings.isNullOrEmpty(record.getArticleIds())) {				
				String[] articles = record.getArticleIds().split(",");
				String[] timestamps = record.getTimestamps().split(",");
				distribution = Maps.newHashMap();
				for (int i = 0; i < articles.length; i++) {
					int articleId = Integer.parseInt(articles[i]);					
					long timestamp = Long.parseLong(timestamps[i]);
					if (category.containsKey(articleId)) {
						int categoryId = category.get(articleId);
						if (distribution.containsKey(categoryId)) {
							int freq = distribution.get(categoryId);
							freq++;
							distribution.put(categoryId, freq);
						} else
							distribution.put(categoryId, 1);
					}
					if (current - timestamp < TWO_DAYS)
						list.add(new UserRecord(articleId, timestamp));
				}
				Collections.sort(list);
				StringBuilder articleSb = new StringBuilder();
				StringBuilder timestampSb = new StringBuilder();
				StringBuilder distributionSb = new StringBuilder();
				int count = 0;
				for (UserRecord ur: list) {
					int articleId = ur.getArticleId();
					if (category.containsKey(articleId)) {
						int categoryId = category.get(articleId);
						if (distribution.containsKey(categoryId)) {
							int freq = distribution.get(categoryId);
							freq--;
							if (freq > 0)
								distribution.put(categoryId, freq);
							else
								distribution.remove(categoryId);
						} 
					}					
					articleSb.append(articleId).append(",");
					timestampSb.append(ur.getTimestamp()).append(",");
					count++;
					if (count == 15)
						break;
				}
				if (distribution.keySet().size() > 0) {
					for (Map.Entry<Integer, Integer> entry: distribution.entrySet()) {
						distributionSb.append(entry.getKey()).append(",").append(entry.getValue()).append(" ");
					}
				}
				list = Lists.newArrayList();
				byte[] row = Bytes.toBytes(userId);
				if (articleSb.length() > 0) {
					StringUtils.deleteLastChar(articleSb);
					StringUtils.deleteLastChar(timestampSb);
					byte[] articleColValue = Bytes.toBytes(articleSb.toString());
					byte[] timestampColValue = Bytes.toBytes(timestampSb.toString());
					Put put = new Put(row);
					put.add(TempUserTable.FAM_MAIN, TempUserTable.COL_VISITED_ARTICLES, articleColValue);
					put.add(TempUserTable.FAM_MAIN, TempUserTable.COL_VISITED_TIMESTAMPS, timestampColValue);
					if (distributionSb.length() > 0) {
						StringUtils.deleteLastChar(distributionSb);
						byte[] distributionColValue = Bytes.toBytes(distributionSb.toString());
						put.add(TempUserTable.FAM_MAIN, TempUserTable.COL_DISTRIBUTION, distributionColValue);
					}
					puts.add(put);
				} else {
					Delete del = new Delete(row);
					deletes.add(del);
				}
			}			
		}
		if (puts.size() > 0)
			table.put(puts);
		if (deletes.size() > 0)
			table.delete(deletes);
	}
	
	private static void cutOff() throws IOException {
		HTable table = new HTable(UserTable.NAME);                
		HTable temp = new HTable(TempUserTable.NAME);
		Map<String, UserMultiRecord> map = Maps.newHashMap();
		try {
			int count = 0;
			Scan scan = new Scan();
			ResultScanner rs = table.getScanner(scan);
			for (Result r : rs) {
				count++;
				String userId = Bytes.toString(r.getRow());
				String articleIds = TempUserTable.getVisitedArticles(r);
				String timestamps = TempUserTable.getVisitedTimestamps(r);
				map.put(userId, new UserMultiRecord(articleIds, timestamps));
				if (count == MAX_BLOCK_SIZE) {
					cutOff(temp, map);
					map = Maps.newHashMap();
					count = 0;
				}
			}
			if (count > 0)
				cutOff(temp, map);
		} catch (Exception e) {
			System.err.println("Error while process cutoff");
		} finally {
			table.close();
			temp.close();
		}
	}
	
	public static void concat() throws IOException {
		HTable temp = new HTable(TempUserTable.NAME);
		HTable table = new HTable(UserTable.NAME);                
        Map<String, UserMultiRecord> map = Maps.newHashMap();        
		try {
			int count = 0;
			Scan scan = new Scan();
            ResultScanner rs = temp.getScanner(scan);
            for (Result r:rs) {
            	count++;
            	String userId = Bytes.toString(r.getRow());
            	String articleIds = TempUserTable.getVisitedArticles(r);
            	String timestamps = TempUserTable.getVisitedTimestamps(r);
            	map.put(userId, new UserMultiRecord(articleIds, timestamps));
            	if (count == MAX_BLOCK_SIZE) {
            		process(table, map);
            		map = Maps.newHashMap();
            		count = 0;
            	}            	
            }
            if (count > 0)
            	process(table, map);
		} catch (Exception e) {
			System.err.println("Error while process concat");
		} finally {
			temp.close();
			table.close();
			try {
				loadCategory();
				TableUtils.truncateTable(TempUserTable.NAME_BYTES);
				cutOff();
			} catch (Exception ex) {
				
			}
		}		
	}
	
	public static void main(String[] args) throws IOException {
		HTable temp = new HTable(TempUserTable.NAME);
		byte[] row = Bytes.toBytes("test");
		Delete del = new Delete(row);
		temp.delete(del);
		temp.close();
		
	}

}
